package com.cennavi.funcs;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;

import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月28日 下午7:42:18 
    * @version 1.0 
    */

public class DeriveFlowFromSpeedPairFunction implements PairFunction<Tuple2<String, String>, String, String> {

	private static final long serialVersionUID = 1L;

	private HashMap<String,String> rtic_kindMap;
	private HashMap<String,String> rtic_freeFlowSpeedMap;
	private HashMap<String,String> rtic_laneNumMap;
	
	public DeriveFlowFromSpeedPairFunction(
				Broadcast<HashMap<String,String>> rtic_kind_broadMap,       
				Broadcast<HashMap<String,String>> rtic_freeFlowSpeedbroadMap,
				Broadcast<HashMap<String,String>> rtic_laneNumMap
			){
				this.rtic_kindMap = rtic_kind_broadMap.value();
				this.rtic_freeFlowSpeedMap = rtic_freeFlowSpeedbroadMap.value();
				this.rtic_laneNumMap = rtic_laneNumMap.value();
			}
	
	public Tuple2<String, String> call(Tuple2<String, String> arg0) throws Exception {
		// 根据速度和匹配数据计算流量
		//********************流量计算相关参数************************
	    //自由流速度
	    double vf = 0.0;
	    //路段速度（模型输入速度）
	    double v = 0.0;
	    //流量
	    int flow = 0;
	    //跟驰速度
	    final double vq = 20.0 / 3.6;
	    //刹车安全系数
	    double a = 0.0;
	    //加速度
	    final double a2 = 0.5;
	    //反应时间（秒）
	    final double b = 1.0;
	    //车身长度与停车安全距离（米）
	    final double c = 9.0;
	    //********************流量计算相关参数************************
	    
	    //********************输出相关********************
	    //输出的字段依次为
	    String cityCode = "610100";
	    String duration = "2";
	    String type = "2";
	    String method = "2";
	    String speed = "null";
	    String trafficIndex = "null";
	    String saturation = "null";
	    String capacity = "null";
	    String value = "";
	    //********************输出相关********************
		String[] data = arg0._2.split(",");
		//速度从km/h转换为m/s，那么算出的流量也是 多少辆/s，如果15分钟流量，就乘以900
		v = Double.parseDouble(data[6]) / 3.6;
		
//		if(rtic_freeFlowSpeedMap.containsKey(data[0]) && rtic_laneNumMap.containsKey(data[0])){
		if(rtic_freeFlowSpeedMap.containsKey(data[0])){
			//邱姐询问认为t_roadsection中的自由流速度可用，遂改为直接从其中获取
			vf = Double.parseDouble(rtic_freeFlowSpeedMap.get(data[0])) / 3.6;
			if(vf < vq || vf == vq){
				vf = vq + 1;
			}
			//计算流量
			a = (v - vq) / (2 * a2 * (vf - vq));
			//输出行先后追加 citycode timestamp duration type method	id 
			//speed traffic_index flow	saturation capacity
			value += cityCode + ",";
			//value += data[1] + ",";
			//value += DateUtil.get5min(new Date()) + ",";
			value += new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + ",";
			value += duration + ",";
			value += type + ",";
			value += method + ",";
			value += data[0] + ",";
			//value += rtic_kindMap.get(data[0]) + ",";
			value += data[0].substring(6, 7) + ",";
			value += speed + ",";
			value += trafficIndex + ",";
			if(v > vf || v == vf){
				v = vf - 1;
				flow = (int)Math.ceil(v / (a * v * v + b * v + c) * 900);
			}else if(v > vq && v < vf){
				flow = (int)Math.ceil(v / (a * v * v + b * v + c) * 900);
			}else{
				flow = (int)Math.ceil(v / (b * v + c) * 900);
			}
			
			//**********单车道流量反推值乘以车道数***********
			//flow = flow * Integer.parseInt(rtic_laneNumMap.get(data[0]));
			
			//**********根据道路类型进行路口流量折减***********
			//reason：rtic速度是剔除了路口等待时间计算出的速度，
			//而通行能力已经进行了路口折减，如果流量计算不进行折减，则可能出现饱和度大于1的现象，与理论情况不符
//			if(null != rtic_kindMap.get(data[0])){
//				if(rtic_kindMap.get(data[0]).equals("5")){
//					flow = (int)Math.floor(flow * 0.7);
//				}else if(rtic_kindMap.get(data[0]).equals("4")
//						|| rtic_kindMap.get(data[0]).equals("6")
//						|| rtic_kindMap.get(data[0]).equals("7")){
//					flow = (int)Math.floor(flow * 0.6);
//				}
//			}
			if(data[0].substring(6, 7).equals("5")){
				flow = (int)Math.floor(flow * 0.7);
			}else if(data[0].substring(6, 7).equals("4")
					|| data[0].substring(6, 7).equals("6")
					|| data[0].substring(6, 7).equals("7")){
				flow = (int)Math.floor(flow * 0.6);
			}
//			value += new DecimalFormat("######0.00").format(v) + ",";
//			value += new DecimalFormat("######0.00").format(vf) + ",";
			value += flow;
			return new Tuple2<String, String>(arg0._1,value);
		}else{
			return null;
		}	
		
	}

}
